package com.galeno.day05;

import com.galeno.FlinkUtils;
import com.galeno.FlinkUtils2;
import jdk.nashorn.internal.runtime.regexp.joni.Config;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2021:33
 */
public class ProcessionTimeNoKeyBy {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",10086);
        StreamExecutionEnvironment env = FlinkUtils.getEnv();
        DataStreamSource<String> lines = env.socketTextStream("192.168.77.101", 8888);
        SingleOutputStreamOperator<Integer> numed = lines.map(Integer::parseInt);
        //AllWindowedStream<Integer, TimeWindow> windowedStream = numed.timeWindowAll(Time.seconds(5));
        AllWindowedStream<Integer, TimeWindow> windowedStream1 = numed.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)));

        SingleOutputStreamOperator<Integer> res = windowedStream1.sum(0);
        res.print();
        env.execute();




    }
}
